package com.wpen.unit.core.service.impl;

import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import com.wpen.unit.commun.BaseResponse;
import com.wpen.unit.core.cache.CacheAgents;
import com.wpen.unit.core.cache.CacheDatas;
import com.wpen.unit.core.ddo.JobAgentBindingDo;
import com.wpen.unit.core.ddo.JobRunningDo;
import com.wpen.unit.core.service.AgentScheduleHandler;
import com.wpen.unit.core.service.CommunityHandler;
import com.wpen.unit.enums.ExecStatus;
import com.wpen.unit.enums.OnlineStatus;
import com.wpen.unit.infra.mybatis.job.po.AgentExec1LogPo;
import com.wpen.unit.infra.mybatis.job.po.AgentRegistryPo;
import com.wpen.unit.infra.mybatis.job.po.JobExec1LogPo;
import com.wpen.unit.infra.mybatis.job.po.JobUndoPlanPo;
import com.wpen.unit.infra.mybatis.job.po.TaskInfoPo;
import com.wpen.unit.infra.repository.job.AgentExec1LogRepository;
import com.wpen.unit.infra.repository.job.AgentRegistryRepository;
import com.wpen.unit.infra.repository.job.JobExec1LogRepository;
import com.wpen.unit.infra.repository.job.JobUndoPlanRepository;
import com.wpen.unit.utils.JsonUtil;
import com.wpen.unit.utils.SpringUtil;
import com.wpen.unit.web.WebVariables;
import com.wpen.unit.web.dto.ContextRequest;
import com.wpen.unit.web.dto.ContextResponse;
import com.wpen.unit.web.utils.HttpService;

import lombok.extern.slf4j.Slf4j;

/**
 * 调度器调度
 *
 * @author Wang Peng
 * @date 2022年12月12日
 * 
 */
@Service
@Slf4j
public class AgentScheduleHandlerImpl implements AgentScheduleHandler {
	@Value("${wpen.application.code.az.mapping}")
	private String azMapping;
	@Value("${wpen.application.agent.service-path}")
	private String agentMonitorPath;

	private static ThreadPoolExecutor agentMonitorPool = null;
	private static ThreadPoolExecutor jobExecutorPool = null;

	static {
		if (agentMonitorPool == null) {
			agentMonitorPool = new ThreadPoolExecutor(4, 8, 2, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(512),
					new CustomizableThreadFactory("agent-monitor-pool-%d"), new ThreadPoolExecutor.CallerRunsPolicy());
			jobExecutorPool = new ThreadPoolExecutor(4, 8, 2, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(8),
					new CustomizableThreadFactory("job-executor-pool-%d"), new ThreadPoolExecutor.CallerRunsPolicy());
		}
	}

	@Autowired
	private CacheAgents cacheAgents;
	@Autowired
	private CacheDatas cacheDatas;
	@Autowired
	private JobUndoPlanRepository jobUndoPlanRepository;
	@Autowired
	private JobExec1LogRepository jobExec1LogRepository;
	@Autowired
	private AgentExec1LogRepository agentExec1LogRepository;
	@Autowired
	private AgentRegistryRepository agentRegistryRepository;
	@Autowired
	private CommunityHandler communityHandler;

	// 未执行的job, key=az-jarname;
	private static Map<String, List<JobUndoPlanPo>> undoPlansMap = new ConcurrentHashMap<>();
	// 内存中的数据，key=job-id， value=状态
	private static Map<Long, String> memoryKeys = new ConcurrentHashMap<>();
	// 执行中的job, key=az; value=[key=ipv4
	private static Map<String, List<JobRunningDo>> runningJobs = new ConcurrentHashMap<>();
	private static Map<Long, JobRunningDo> runningStatus = new HashMap<>();
	private static Map<Long, Integer> runningCheckNums = new HashMap<>();

	@Override
	public void agentMonitor() {
		List<AgentRegistryPo> agentRegistryPos = cacheAgents.getAgentsByAz();
		if (log.isDebugEnabled()) {
			log.debug("Agent Monitor, Agent>" + (agentRegistryPos == null ? 0 : agentRegistryPos.size()));
		}
		if (agentRegistryPos == null) {
			return;
		}
		// 如果az中存在运行中的任务，直接使用结果查询接口替换探针接口，如果结果查询失败，再使用探针
		Map<String, List<JobRunningDo>> agentRuns = runningJobs;
		for (AgentRegistryPo agentPo : agentRegistryPos) {
			// 多线程并行
			boolean health = false;
			if (agentRuns == null || agentRuns.isEmpty()) {
				health = true;
			} else {
				List<JobRunningDo> runningDos = agentRuns.get(agentPo.getAgentIpv4());
				if (runningDos == null || runningDos.isEmpty()) {
					health = true;
				}
			}
			if (health) {
				agentMonitorPool.execute(new Runnable() {

					@Override
					public void run() {
						boolean ok = false;
						for (int i = 0; i < 3; i++) {
							try {
								// 发送http请求健康检测
								BaseResponse baseResponse = http2Health(agentPo);
								if (WebVariables.RESPONSE_SUCC_CODE.equals(baseResponse.getCode())) {
									ok = true;
									break;
								}
							} catch (Exception e) {
								// 请求不通了，重试3次
								try {
									TimeUnit.MILLISECONDS.sleep(100);
								} catch (InterruptedException e1) {
									// ignore
								}
							}
						}
						if (!ok) {
							log.info("Agent Monitor ... IP>" + agentPo.getAgentIpv4() + " Health Fail !!!");
							int failNum = cacheAgents.check2Fail(agentPo.getAgentIpv4());
							if (failNum > 2) {
								// 不在线了
							}
						} else {
							cacheAgents.check2Online(agentPo.getAgentIpv4());
						}
					}
				}); // end thread
				continue;
			}
			List<JobRunningDo> runningDos = agentRuns.get(agentPo.getAgentIpv4());
			if (runningDos == null || runningDos.isEmpty()) {
				continue;
			}
			for (int i = 0; i < runningDos.size(); i++) {
				JobRunningDo jobRunningDo = runningDos.get(i);
				if (!agentPo.getAgentIpv4().equals(jobRunningDo.getIpv4())) {
					continue;
				}
				boolean toCheck = true;
				if (ExecStatus.Init.value().equals(jobRunningDo.getAgentExeStatus())) {
					log.info(">>===" + jobRunningDo.getIpv4() + "-" + jobRunningDo.getJobLogId() + "=Status[ "
							+ jobRunningDo.getAgentExeStatus() + " ] Discrepancy !! ");
					Integer checkNum = 1;
					if (runningCheckNums.containsKey(jobRunningDo.getJobLogId())) {
						checkNum = runningCheckNums.get(jobRunningDo.getJobLogId());
						runningCheckNums.put(jobRunningDo.getJobLogId(), checkNum + 1);
					} else {
						runningCheckNums.put(jobRunningDo.getJobLogId(), checkNum);
					}
					// 7345 已经处在undo-R，agent-0，非正常的状态上了。5s中检测一次，12次是1分钟
					// 24 次等待，如果还是没有正常初始化，强行检测，结果是NOTFOUND异常
					if (checkNum >= 24) {
						jobRunningDo.setAgentExeStatus(ExecStatus.Running.value());
					}
					toCheck = false;
				}
				if (toCheck) {
					log.info(">>===Running>>size: " + runningDos.size() + ". ==>" + agentPo.getAgentIpv4()
							+ " Exec Http Check.");
				}
				agentMonitorPool.execute(new Runnable() {

					@Override
					public void run() {
						if (jobRunningDo != null && !ExecStatus.Init.value().equals(jobRunningDo.getAgentExeStatus())) {
							try {
								// 检测任务是否执行成功
								ContextResponse response = http2Check(agentPo, jobRunningDo);
								// 成功了
								communityHandler.checkFinished(jobRunningDo, response);
								if (WebVariables.RESPONSE_SUCC_CODE.equals(response.getCode())
										|| WebVariables.RESPONSE_AGENT_NOTFOUND.equals(response.getCode())
										|| WebVariables.RESPONSE_AGENT_EXCEPTION.equals(response.getCode())) {
									removeRunnings(agentPo.getAgentIpv4(), jobRunningDo);
									removeUndoPlans(jobRunningDo.getAppJarName(), jobRunningDo.getJobLogId(),
											agentPo.getAgentIpv4());
									// 删除agent中的任务
									http2Remove(agentPo, jobRunningDo);
								}
								return;
							} catch (Exception e) {
								log.info("Agent Monitor ... IP>" + agentPo.getAgentIpv4() + " Check Fail !!!");
							}
						}
						boolean ok = false;
						for (int i = 0; i < 3; i++) {
							try {
								// 发送http请求健康检测
								BaseResponse baseResponse = http2Health(agentPo);
								if (WebVariables.RESPONSE_SUCC_CODE.equals(baseResponse.getCode())) {
									ok = true;
									break;
								}
							} catch (Exception e) {
								// 请求不通了，重试3次
								try {
									TimeUnit.MILLISECONDS.sleep(100);
								} catch (InterruptedException e1) {
									// ignore
								}
							}
						}
						if (!ok) {
							log.info("Agent Monitor ... IP>" + agentPo.getAgentIpv4()
									+ " Health Fail !!!");
							int failNum = cacheAgents.check2Fail(agentPo.getAgentIpv4());
							if (failNum > 2) {
								// agent已宕机
								ContextResponse response = new ContextResponse();
								response.setCode(WebVariables.RESPONSE_AGENT_EXCEPTION);
								response.setMsg("Agent长时间无响应，异常失败");
								response.setEndDate(new Date());
								response.setExitValue(1);
								response.setStatus(ExecStatus.Excption.value());
								communityHandler.checkFinished(jobRunningDo, response);
								removeRunnings(agentPo.getAgentIpv4(), jobRunningDo);
								removeUndoPlans(jobRunningDo.getAppJarName(), jobRunningDo.getJobLogId(),
										agentPo.getAgentIpv4());
							}
						} else {
							cacheAgents.check2Online(agentPo.getAgentIpv4());
						}
					}
				}); // end thread
			} // end undo
		} // end for
	}

	public ContextResponse http2Check(AgentRegistryPo agentPo, JobRunningDo jobRunningDo) {
		HttpService<ContextResponse> httpService = new HttpService<>();
		String port = agentPo.getAgentPort() == null ? "" : ":" + agentPo.getAgentPort();
		String ipAddress = agentPo.getAgentIpv4() + port;
		String servicePath = agentMonitorPath.replaceAll("\\{ip_address\\}", ipAddress);
		ContextRequest contextRequest = new ContextRequest();
		contextRequest.setBizCode(jobRunningDo.getBizCode());
		contextRequest.setBatchCode(jobRunningDo.getBatchCode());
		contextRequest.setTaskCode(jobRunningDo.getTaskCode());
		contextRequest.setAcctDate(jobRunningDo.getAcctDate());
		contextRequest.setTaskId(jobRunningDo.getTaskLogId());
		contextRequest.setJobId(jobRunningDo.getJobLogId());
		contextRequest.setCmd("check");
		contextRequest.setParams("");
		ContextResponse response = httpService.proxyPostHttp(servicePath, WebVariables.URI_AGENT_CHECK, null,
				contextRequest, ContextResponse.class, false);
		return response;
	}

	public ContextResponse http2Remove(AgentRegistryPo agentPo, JobRunningDo jobRunningDo) {
		HttpService<ContextResponse> httpService = new HttpService<>();
		String port = agentPo.getAgentPort() == null ? "" : ":" + agentPo.getAgentPort();
		String ipAddress = agentPo.getAgentIpv4() + port;
		String servicePath = agentMonitorPath.replaceAll("\\{ip_address\\}", ipAddress);
		ContextRequest contextRequest = new ContextRequest();
		contextRequest.setBizCode(jobRunningDo.getBizCode());
		contextRequest.setBatchCode(jobRunningDo.getBatchCode());
		contextRequest.setTaskCode(jobRunningDo.getTaskCode());
		contextRequest.setAcctDate(jobRunningDo.getAcctDate());
		contextRequest.setTaskId(jobRunningDo.getTaskLogId());
		contextRequest.setJobId(jobRunningDo.getJobLogId());
		contextRequest.setCmd("remove");
		contextRequest.setParams("");
		ContextResponse response = httpService.proxyPostHttp(servicePath, WebVariables.URI_AGENT_REMOVE, null,
				contextRequest, ContextResponse.class, false);
		return response;
	}

	public ContextResponse http2Kill(AgentRegistryPo agentPo, JobRunningDo jobRunningDo) {
		HttpService<ContextResponse> httpService = new HttpService<>();
		String port = agentPo.getAgentPort() == null ? "" : ":" + agentPo.getAgentPort();
		String ipAddress = agentPo.getAgentIpv4() + port;
		String servicePath = agentMonitorPath.replaceAll("\\{ip_address\\}", ipAddress);
		ContextRequest contextRequest = new ContextRequest();
		contextRequest.setBizCode(jobRunningDo.getBizCode());
		contextRequest.setBatchCode(jobRunningDo.getBatchCode());
		contextRequest.setTaskCode(jobRunningDo.getTaskCode());
		contextRequest.setAcctDate(jobRunningDo.getAcctDate());
		contextRequest.setTaskId(jobRunningDo.getTaskLogId());
		contextRequest.setJobId(jobRunningDo.getJobLogId());
		contextRequest.setCmd("kill");
		contextRequest.setParams("");
		ContextResponse response = httpService.proxyPostHttp(servicePath, WebVariables.URI_AGENT_KILL, null,
				contextRequest, ContextResponse.class, false);
		return response;
	}

	private BaseResponse http2Health(AgentRegistryPo agentPo) {
		HttpService<BaseResponse> httpService = new HttpService<>();
		String port = agentPo.getAgentPort() == null ? "" : ":" + agentPo.getAgentPort();
		String ipAddress = agentPo.getAgentIpv4() + port;
		String servicePath = agentMonitorPath.replaceAll("\\{ip_address\\}", ipAddress);
		BaseResponse baseResponse = httpService.proxyGetHttp(servicePath, WebVariables.URI_AGENT_HEALTH, null, null,
				BaseResponse.class, false);
		return baseResponse;
	}

	private void addUndoPlans(JobUndoPlanPo undo) {
		memoryKeys.put(undo.getId(), undo.getExecStatus());
		String undoKey = undo.getAppJarName();
		if (undoPlansMap.containsKey(undoKey)) {
			List<JobUndoPlanPo> undoLists = undoPlansMap.get(undoKey);
			boolean undoHas = false;
			for (JobUndoPlanPo jobUndoPlanPo : undoLists) {
				if (Objects.equals(undo.getId(), jobUndoPlanPo.getId())) {
					undoHas = true;
					break;
				}
			}
			if (!undoHas) {
				undoLists.add(undo.clone());
			}
		} else {
			List<JobUndoPlanPo> undoLists = new LinkedList<>();
			undoLists.add(undo.clone());
			undoPlansMap.put(undoKey, undoLists);
		}
	}

	private void removeUndoPlans(String jarName, Long jobId, String ipv4) {
		String undoKey = jarName;
		if (undoPlansMap.containsKey(undoKey)) {
			List<JobUndoPlanPo> undoLists = undoPlansMap.get(undoKey);
			if (undoLists == null) {
				return;
			}
			for (JobUndoPlanPo jobUndoPlanPo : undoLists) {
				if (Objects.equals(jobId, jobUndoPlanPo.getId())) {
					// 清理资源
					// 清理记忆
					undoLists.remove(jobUndoPlanPo);
					memoryKeys.remove(jobId);
					return;
				}
			}
		}
	}

	private void removeRunnings(String ipv4, JobRunningDo jobRunningDo) {
		List<JobRunningDo> runningDos = runningJobs.get(ipv4);
		if (runningDos != null) {
			for (JobRunningDo jobRunningDo2 : runningDos) {
				if (Objects.equals(jobRunningDo2.getJobLogId(), jobRunningDo.getJobLogId())) {
					runningDos.remove(jobRunningDo2);
					runningStatus.remove(jobRunningDo2.getJobLogId());
					runningCheckNums.remove(jobRunningDo2.getJobLogId());
					return;
				}
			}

		}
	}

	private synchronized void addRunnings(String ipv4, JobRunningDo jobRunningDo) {
		List<JobRunningDo> runningDos = runningJobs.get(ipv4);
		if (runningDos != null) {
			boolean undoHas = false;
			for (JobRunningDo JobRunningDo2 : runningDos) {
				if (Objects.equals(JobRunningDo2.getJobLogId(), jobRunningDo.getJobLogId())) {
					undoHas = true;
					break;
				}
			}
			if (!undoHas) {
				runningDos.add(jobRunningDo);
				runningStatus.put(jobRunningDo.getJobLogId(), jobRunningDo);
			}
		} else {
			runningDos = new ArrayList<>();
			runningDos.add(jobRunningDo);
			runningJobs.put(ipv4, runningDos);
			runningStatus.put(jobRunningDo.getJobLogId(), jobRunningDo);
		}

	}

	@Override
	// 不要加事务注解
	public void jobExecutor() {
		int maxCache = 16;
		String[] azStrs = azMapping.split(",");
		Set<String> azs = new HashSet<>();
		for (String azConf : azStrs) {
			azs.add(azConf.substring(azConf.indexOf("=") + 1));
		}
		if (azs == null || azs.isEmpty()) {
			return;
		}
		for (String az : azs) {
			// 查询所有待执行作业
			List<JobUndoPlanPo> undoPlanPos = jobUndoPlanRepository.queryListByTriggerTime(maxCache, az);
			int undoSize = 0;
			if (undoPlanPos != null) {
				undoSize = undoPlanPos.size();
				for (JobUndoPlanPo undo : undoPlanPos) {
					addUndoPlans(undo);
				}
			}
			if (log.isDebugEnabled()) {
				log.debug("Job Executor>> memory size>" + memoryKeys.size() + ", AZ[" + az + "] query form db >"
						+ undoSize);
			}
		}
		if (memoryKeys == null || memoryKeys.isEmpty()) {
			return;
		}
		for (String key : undoPlansMap.keySet()) {
			// 对区域-jar进行循环，查询区域下所有待执行任务
			List<JobUndoPlanPo> jobUndoPlanPos = undoPlansMap.get(key);
			if (jobUndoPlanPos != null && !jobUndoPlanPos.isEmpty()) {
				List<JobUndoPlanPo> innerUndoPlanPos = new ArrayList<>();

				for (JobUndoPlanPo undoPlanPo : jobUndoPlanPos) {
					if (ExecStatus.Init.value().equals(undoPlanPo.getExecStatus())) {
						innerUndoPlanPos.add(undoPlanPo.clone());
					} else if (ExecStatus.Running.value().equals(undoPlanPo.getExecStatus())) {
						if (runningStatus.containsKey(undoPlanPo.getId())) {
							continue;
						} else {
							innerUndoPlanPos.add(undoPlanPo.clone());
						}
					}
				}
				// 如果innerUndoPlanPos为空了，就
				if (innerUndoPlanPos == null || innerUndoPlanPos.isEmpty()) {
					continue;
				}

				// 悲观锁更新库表，，更新成功后，远程调用
				AgentScheduleHandler agentScheduleHandler = SpringUtil.getBean(AgentScheduleHandler.class);
				JobAgentBindingDo agentJobBindingDo = agentScheduleHandler.jobExecutorUpdate(innerUndoPlanPos);
				if (agentJobBindingDo == null) {
					continue;
				}
				// 远程调用
				AgentRegistryPo agentPo = agentJobBindingDo.getAgentRegistryPo();
				JobExec1LogPo jobExec1LogPo = agentJobBindingDo.getJobExec1LogPo();
				AgentExec1LogPo agentExec1LogPo = agentJobBindingDo.getAgentExec1LogPo();
				JobUndoPlanPo jobUndoPlanPo = agentJobBindingDo.getJobUndoPlanPo();

				JobRunningDo jobRunningDo = new JobRunningDo();
				//
				jobRunningDo.setBizCode(jobExec1LogPo.getBizCode());
				jobRunningDo.setBatchCode(jobExec1LogPo.getBatchCode());
				jobRunningDo.setTaskCode(jobExec1LogPo.getTaskCode());
				jobRunningDo.setAcctDate(jobExec1LogPo.getExecDate());
				jobRunningDo.setAppJarName(jobUndoPlanPo.getAppJarName());
				//
				jobRunningDo.setAppMode(jobUndoPlanPo.getAppMode());
				jobRunningDo.setTaskLogId(jobExec1LogPo.getTaskExecLogId());
				jobRunningDo.setJobLogId(jobUndoPlanPo.getId());
				jobRunningDo.setAgentId(agentExec1LogPo.getId());
				jobRunningDo.setIpv4(agentPo.getAgentIpv4());
				jobRunningDo.setJobTriggerTime(jobUndoPlanPo.getJobTriggerTime());
				jobRunningDo.setUndoExecStatus(jobUndoPlanPo.getExecStatus());
				jobRunningDo.setAgentExeStatus(agentExec1LogPo.getExecStatus());
				// 添加到运行中
				addRunnings(agentPo.getAgentIpv4(), jobRunningDo);

				for (JobUndoPlanPo rmUndoPlanPo : jobUndoPlanPos) {
					if (jobUndoPlanPo.getId().equals(rmUndoPlanPo.getId())) {
						memoryKeys.remove(rmUndoPlanPo.getId());
						jobUndoPlanPos.remove(rmUndoPlanPo);
						break;
					}
				}
				// 多线程调度
				jobExecutorPool.execute(new Runnable() {

					@Override
					public void run() {
						try {
							HttpService<ContextResponse> httpService = new HttpService<>();
							String port = agentPo.getAgentPort() == null ? "" : ":" + agentPo.getAgentPort();
							String ipAddress = agentPo.getAgentIpv4() + port;
							String servicePath = agentMonitorPath.replaceAll("\\{ip_address\\}", ipAddress);
							ContextRequest contextRequest = new ContextRequest();
							contextRequest.setBizCode(jobExec1LogPo.getBizCode());
							contextRequest.setBatchCode(jobExec1LogPo.getBatchCode());
							contextRequest.setTaskCode(jobExec1LogPo.getTaskCode());
							contextRequest.setAcctDate(jobExec1LogPo.getExecDate());
							contextRequest.setTaskId(jobExec1LogPo.getTaskExecLogId());
							contextRequest.setJobId(jobExec1LogPo.getId());
							contextRequest.setCmd(agentExec1LogPo.getExecStartupCmd());
							String params = jobExec1LogPo.getExecParams();
							if (params != null) {
								params = params.replaceAll("\r\n", "").replaceAll("\n", "");
							}
							contextRequest.setParams(params);
							ContextResponse response = httpService.proxyPostHttp(servicePath,
									WebVariables.URI_AGENT_START, null, contextRequest, ContextResponse.class, false);

							// 这个接口返回成功
							if (WebVariables.RESPONSE_SUCC_CODE.equals(response.getCode())) {
								// 成功了 ，， 更新agent状态 运行中
								if (runningStatus.containsKey(agentExec1LogPo.getJobExecLogId())) {
									runningStatus.get(agentExec1LogPo.getJobExecLogId())
											.setAgentExeStatus(ExecStatus.Running.value());
								}
								AgentExec1LogPo upAgentExec1Log = new AgentExec1LogPo();
								upAgentExec1Log.setId(agentExec1LogPo.getId());
								upAgentExec1Log.setExecStatus(ExecStatus.Running.value());
								agentExec1LogRepository.updateById(upAgentExec1Log);
							} else {
								// 已经明确异常了，直接失败
								response.setCode(WebVariables.RESPONSE_AGENT_EXCEPTION);
								response.setStatus(ExecStatus.Excption.value());
								communityHandler.checkFinished(jobRunningDo, response);
								// 删除running中数据
								removeRunnings( agentPo.getAgentIpv4(), jobRunningDo);
								removeUndoPlans( jobRunningDo.getAppJarName(), jobRunningDo.getJobLogId(),
										jobRunningDo.getIpv4());
							}
						} catch (Exception e) {
							// 连接异常，重新；；；等待异常，去检测；；；其他异常直接失败
							if (e instanceof SocketTimeoutException || e.getCause() instanceof SocketTimeoutException) {
								// 创建连接超时，任务无法执行
								String key = jobExec1LogPo.getBizCode() + "-" + jobExec1LogPo.getBatchCode() + "-"
										+ jobExec1LogPo.getTaskCode() + "-" + jobExec1LogPo.getExecDate() + "-"
										+ jobExec1LogPo.getId();
								// 更新agent日志失败记录次数，如果达到3次失败，这个机器不可用了，自动清理agent日志，undo会ready状态
								log.warn("Failed to send job[" + key + "] to server " + agentPo.getAgentIpv4() + "");
								// 故障转移
								removeRunnings( agentPo.getAgentIpv4(), jobRunningDo);
								communityHandler.updateAfterSendSocketTimeoutException(jobRunningDo);
							}
							// 其他异常不明确了，， 交给检测程序
							// 如果agent响应超时，错误 java.net.SocketTimeoutException: timeout
							// 如果agent直接kill了，错误 SocketException: Connection reset by peer
							// 如果agent跑死了，错误 Caused by: java.io.IOException: unexpected end of stream on
							log.error("jobExecutor() >" + JsonUtil.toJson(jobUndoPlanPo), e);
						}
					}
				});
			}
		}
	}

	@Transactional(propagation = Propagation.REQUIRES_NEW)
	public JobAgentBindingDo jobExecutorUpdate(List<JobUndoPlanPo> jobUndoPlanPos) {
		int size = jobUndoPlanPos.size();
		for (int i = 0; i < size; i++) {
			JobUndoPlanPo jobUndoPlanPo = jobUndoPlanPos.remove(0);
			if (jobUndoPlanPo == null) {
				return null;
			}
			JobUndoPlanPo jobUndoPlanPoLock = jobUndoPlanRepository.queryByIdWithLock(jobUndoPlanPo.getId());
			if (jobUndoPlanPoLock == null) {
				removeUndoPlans( jobUndoPlanPo.getAppJarName(), jobUndoPlanPo.getId(), null);
				continue;
			}
			JobExec1LogPo jobExec1LogPo = jobExec1LogRepository.queryById(jobUndoPlanPo.getId());

			if (ExecStatus.Init.value().equals(jobUndoPlanPoLock.getExecStatus())) {
				// 准备中，获取agent
				// 只有ready状态时需要自己驱动执行的
				TaskInfoPo taskInfoPo = cacheDatas.getTaskInfoByCode(jobExec1LogPo.getBizCode(),
						jobExec1LogPo.getBatchCode(), jobExec1LogPo.getTaskCode());
				// 准备中，获取agent
				AgentRegistryPo agentRegistryPo = getOne( jobExec1LogPo.getBizCode(), taskInfoPo.getAppJarName());
				if (agentRegistryPo == null) {
					// 没有可用的执行器
					String jobStr = jobExec1LogPo.getBatchCode() + "-" + jobExec1LogPo.getTaskCode() + "-"
							+ jobExec1LogPo.getJobCode();
					log.warn("Job>" + jobStr + ", Agent busy Or Not found !!! ");
					return null;
				}
				// 可以执行，，，更新和插入表
				JobAgentBindingDo agentJobBindingDo = communityHandler.updateBeforeSend(jobUndoPlanPoLock,
						jobExec1LogPo, agentRegistryPo, taskInfoPo);
				agentJobBindingDo.setAgentRegistryPo(agentRegistryPo);
				return agentJobBindingDo;
			} else {
				// 其他所有状态，都去检测最终状态
				AgentExec1LogPo agentExec1LogPo = agentExec1LogRepository.queryByJobId(jobUndoPlanPo.getId(),
						jobExec1LogPo.getTaskTriggerTime());
				JobRunningDo jobRunningDo = new JobRunningDo();
				//
				jobRunningDo.setBizCode(jobExec1LogPo.getBizCode());
				jobRunningDo.setBatchCode(jobExec1LogPo.getBatchCode());
				jobRunningDo.setTaskCode(jobExec1LogPo.getTaskCode());
				jobRunningDo.setAcctDate(jobExec1LogPo.getExecDate());
				jobRunningDo.setAppJarName(jobUndoPlanPoLock.getAppJarName());
				//
				jobRunningDo.setAppMode(jobUndoPlanPo.getAppMode());
				jobRunningDo.setTaskLogId(jobExec1LogPo.getTaskExecLogId());
				jobRunningDo.setJobLogId(jobUndoPlanPo.getId());
				jobRunningDo.setAgentId(agentExec1LogPo.getId());
				jobRunningDo.setIpv4(agentExec1LogPo.getAgentIpv4());
				jobRunningDo.setJobTriggerTime(jobUndoPlanPo.getJobTriggerTime());
				jobRunningDo.setUndoExecStatus(jobUndoPlanPo.getExecStatus());
				jobRunningDo.setAgentExeStatus(agentExec1LogPo.getExecStatus());

				// 添加到运行中
				addRunnings(agentExec1LogPo.getAgentIpv4(), jobRunningDo);
				continue;
			}
		}
		return null;
	}

	private AgentRegistryPo getOne(String bizCode, String jarName) {
		int num = 0;
		Set<String> excludeIps = new HashSet<>();
		while (true) {
			num++;
			if (num > 20) {
				log.info("getOne cycled 20 times, forced exit.");
				return null;
			}
			AgentRegistryPo agentOne = cacheAgents.getAgentsCanUse( bizCode,  excludeIps);
			if (agentOne == null) {
				return null;
			}
			String ipv4 = agentOne.getAgentIpv4();
			AgentRegistryPo agentRegistryPo = agentRegistryRepository.queryByIpv4WithLock(ipv4);
			if (OnlineStatus.Offline.value().equals(agentRegistryPo.getOnlineStatus())) {
				cacheAgents.check2Fail(ipv4);
				continue;
			}
			List<AgentExec1LogPo> agentRunnings = agentExec1LogRepository.queryByIpv4ForScore(ipv4);
			if (agentRunnings == null || agentRunnings.isEmpty()) {
				return agentOne;
			}
			try {
				Thread.sleep(5000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}
